#include <fstream>
#include <random>
#include <thread>

#include <bthread/bthread.h>
#include <glog/logging.h>
#include <gtest/gtest.h>

#include "hawking/indexlib/index_config.pb.h"
#include "hawking/indexlib/indexlib_data.pb.h"
#include "doc/document_creator.h"
#include "segment/mem_segment.h"
#include "segment/disk_segment.h"
#include "util/file_wrapper/file_wrapper.h"

using namespace hawking::indexlib;

std::shared_ptr<IndexlibIndexConfig> MakeIndexConfig() {
    std::shared_ptr<IndexlibIndexConfig> index_config = 
        std::make_shared<IndexlibIndexConfig>();
    int32_t field_id = 0;
    // i8
    auto* field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("i8");
    field->set_field_type(hawking::util::ColumnType::COLUMN_INT8);
    // i32
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("i32");
    field->set_field_type(hawking::util::ColumnType::COLUMN_INT32);
    // i64
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("i64");
    field->set_field_type(hawking::util::ColumnType::COLUMN_INT64);
    // float
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("float");
    field->set_field_type(hawking::util::ColumnType::COLUMN_FLOAT);
    // double
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("double");
    field->set_field_type(hawking::util::ColumnType::COLUMN_DOUBLE);
    // string
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("string");
    field->set_field_type(hawking::util::ColumnType::COLUMN_STRING);
    // i8s
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("i8s");
    field->set_field_type(hawking::util::ColumnType::COLUMN_INT8_LIST);
    // i32s
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("i32s");
    field->set_field_type(hawking::util::ColumnType::COLUMN_INT32_LIST);
    // i64s
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("i64s");
    field->set_field_type(hawking::util::ColumnType::COLUMN_INT64_LIST);
    // floats
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("floats");
    field->set_field_type(hawking::util::ColumnType::COLUMN_FLOAT_LIST);
    // doubles
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("doubles");
    field->set_field_type(hawking::util::ColumnType::COLUMN_DOUBLE_LIST);
    // strings
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("strings");
    field->set_field_type(hawking::util::ColumnType::COLUMN_STRING_LIST);

    // inv1
    auto* inverted_configs = index_config->add_inverted_configs();
    inverted_configs->set_field_id(1);
    inverted_configs->set_field_name("i32");

    // inv2
    inverted_configs = index_config->add_inverted_configs();
    inverted_configs->set_field_id(5);
    inverted_configs->set_field_name("string");

    // pk
    index_config->set_pk_field_id(2);

    return index_config;
}

void MakeMsg(IndexlibMessage* msg) {
    auto* field = msg->add_columns();
    field->set_column_name("i8");
    field->mutable_column_value()->set_int_value(100);

    field = msg->add_columns();
    field->set_column_name("i32");
    field->mutable_column_value()->set_int_value(100000);

    field = msg->add_columns();
    field->set_column_name("i64");
    field->mutable_column_value()->set_int_value(10000000);

    field = msg->add_columns();
    field->set_column_name("float");
    field->mutable_column_value()->set_double_value(0.123);

    field = msg->add_columns();
    field->set_column_name("double");
    field->mutable_column_value()->set_double_value(10.456);

    field = msg->add_columns();
    field->set_column_name("string");
    field->mutable_column_value()->set_bytes_value("hello world");

    field = msg->add_columns();
    field->set_column_name("i8s");
    field->mutable_column_value()->mutable_int_values()->add_values(1);
    field->mutable_column_value()->mutable_int_values()->add_values(2);
    field->mutable_column_value()->mutable_int_values()->add_values(3);

    field = msg->add_columns();
    field->set_column_name("i32s");
    field->mutable_column_value()->mutable_int_values()->add_values(999999);
    field->mutable_column_value()->mutable_int_values()->add_values(1999999);
    field->mutable_column_value()->mutable_int_values()->add_values(2999999);

    field = msg->add_columns();
    field->set_column_name("i64s");
    field->mutable_column_value()->mutable_int_values()->add_values(11999999);
    field->mutable_column_value()->mutable_int_values()->add_values(111999999);
    field->mutable_column_value()->mutable_int_values()->add_values(112999999);

    field = msg->add_columns();
    field->set_column_name("floats");
    field->mutable_column_value()->mutable_double_values()->add_values(0.987);
    field->mutable_column_value()->mutable_double_values()->add_values(0.654);
    field->mutable_column_value()->mutable_double_values()->add_values(0.321);

    field = msg->add_columns();
    field->set_column_name("doubles");
    field->mutable_column_value()->mutable_double_values()->add_values(1000.987);
    field->mutable_column_value()->mutable_double_values()->add_values(2000.654);
    field->mutable_column_value()->mutable_double_values()->add_values(3000.321);

    field = msg->add_columns();
    field->set_column_name("strings");
    field->mutable_column_value()->mutable_bytes_values()->add_values("aaaaa");
    field->mutable_column_value()->mutable_bytes_values()->add_values("hello");
    field->mutable_column_value()->mutable_bytes_values()->add_values("world");
}

void MakeAddMsg(IndexlibMessage* msg) {
    MakeMsg(msg);
    msg->set_cmd(hawking::util::IndexOp::INDEX_RT_ADD);
}

void MakeUpdateMsg(IndexlibMessage* msg) {
    MakeMsg(msg);
    msg->set_cmd(hawking::util::IndexOp::INDEX_RT_UPDATE);
}

void MakeDelMsg(IndexlibMessage* msg) {
    MakeMsg(msg);
    msg->set_cmd(hawking::util::IndexOp::INDEX_RT_DEL);
}

TEST(mem_segment_test, add_and_lookup) {
    std::shared_ptr<IndexlibIndexConfig> index_config = MakeIndexConfig();
    MemSegment mem_segment;
    EXPECT_EQ(true, mem_segment.Init(index_config));
    
    std::shared_ptr<DocumentCreator> document_creator =
        std::make_shared<DocumentCreator>(index_config);

    // add one
    IndexlibMessage message;
    MakeAddMsg(&message);
    std::shared_ptr<NormalDocument> doc = document_creator->CreateNormalDocument(message, 0);
    if (!doc) {
        return;
    }
    doc->SetDocId(0);
    LOG(INFO) << mem_segment.AddDocument(doc);

    // select one
    std::string str = std::to_string(100000);
    uint64_t term_key = HashAlgorithm::HashString64(str.data(), str.length());
    google::protobuf::RepeatedPtrField<std::string> select_fields;
    select_fields.Add("i8");
    select_fields.Add("i32");
    select_fields.Add("strings");
    select_fields.Add("i32s");
    select_fields.Add("floats");
    IndexlibRetrieveResponse resp;
    mem_segment.Query(1, term_key, 1, select_fields, {}, {}, &resp);
    LOG(INFO) << resp.ShortDebugString();
}

TEST(mem_segment_localdata_filter_test, add_and_lookup_and_filter) {
    std::shared_ptr<IndexlibIndexConfig> index_config = MakeIndexConfig();
    MemSegment mem_segment;
    EXPECT_EQ(true, mem_segment.Init(index_config));
    
    std::shared_ptr<DocumentCreator> document_creator =
        std::make_shared<DocumentCreator>(index_config);

    // i8 filter test
    IndexlibMessage message;
    MakeAddMsg(&message);
    std::shared_ptr<NormalDocument> doc =
        document_creator->CreateNormalDocument(message, 0);
    if (!doc) {
        return;
    }
    doc->SetDocId(0);
    LOG(INFO) << mem_segment.AddDocument(doc);

    google::protobuf::RepeatedPtrField<hawking::util::FilterColumn> filter_columns;
    google::protobuf::RepeatedPtrField<hawking::util::FilterColumn> where_columns;

//////////////////////////////////////////////////////////////////////////////////// i8
    /// eq
    hawking::util::FilterColumn* filter_eq = filter_columns.Add();
    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(100);
    filter_eq->set_action(hawking::util::ExpressionOp::EQ);
    auto ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(101);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    /// ne
    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(100);
    filter_eq->set_action(hawking::util::ExpressionOp::NE);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(101);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    /// >,<
    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(99);
    filter_eq->set_action(hawking::util::ExpressionOp::LT);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(101);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(99);
    filter_eq->set_action(hawking::util::ExpressionOp::GT);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(101);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    /// >=,<=
    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(100);
    filter_eq->set_action(hawking::util::ExpressionOp::LE);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(99);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(100);
    filter_eq->set_action(hawking::util::ExpressionOp::GE);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(101);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    // in
    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->mutable_int_values()->add_values(99);
    filter_eq->mutable_column_value()->mutable_int_values()->add_values(101);
    filter_eq->set_action(hawking::util::ExpressionOp::IN);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->mutable_int_values()->add_values(100);
    std::sort(
        filter_eq->mutable_column_value()->mutable_int_values()->mutable_values()->begin(),
        filter_eq->mutable_column_value()->mutable_int_values()->mutable_values()->end());
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    // contain
    filter_eq->set_column_name("i8s");
    filter_eq->mutable_column_value()->set_int_value(5);
    filter_eq->set_action(hawking::util::ExpressionOp::CONTAIN);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->set_int_value(2);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    // intersect
    filter_eq->set_column_name("i8s");
    filter_eq->mutable_column_value()->mutable_int_values()->add_values(10);
    filter_eq->mutable_column_value()->mutable_int_values()->add_values(11);
    filter_eq->set_action(hawking::util::ExpressionOp::INTERSECT);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->mutable_int_values()->add_values(1);
    std::sort(
        filter_eq->mutable_column_value()->mutable_int_values()->mutable_values()->begin(),
        filter_eq->mutable_column_value()->mutable_int_values()->mutable_values()->end());
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

///////////////////////////////////////////////////////////////////////////////// float
    /// eq
    filter_eq->set_column_name("float");
    filter_eq->mutable_column_value()->set_double_value(0.123);
    filter_eq->set_action(hawking::util::ExpressionOp::EQ);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_double_value(0.124);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    // ne
    filter_eq->mutable_column_value()->set_double_value(0.123);
    filter_eq->set_action(hawking::util::ExpressionOp::NE);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->set_double_value(0.124);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    // >,<
    filter_eq->mutable_column_value()->set_double_value(0.123);
    filter_eq->set_action(hawking::util::ExpressionOp::LT);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->set_double_value(0.124);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_double_value(0.123);
    filter_eq->set_action(hawking::util::ExpressionOp::GT);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->set_double_value(0.122);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    // >=,<=
    filter_eq->mutable_column_value()->set_double_value(0.123);
    filter_eq->set_action(hawking::util::ExpressionOp::LE);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_double_value(0.122);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->set_double_value(0.123);
    filter_eq->set_action(hawking::util::ExpressionOp::GE);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_double_value(0.124);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    // in
    float f1 = 0.122, f2 = 0.124, f3 = 0.123;
    filter_eq->mutable_column_value()->mutable_double_values()->add_values(f1);
    filter_eq->mutable_column_value()->mutable_double_values()->add_values(f2);
    filter_eq->set_action(hawking::util::ExpressionOp::IN);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->mutable_double_values()->add_values(f3);
    std::sort(
        filter_eq->mutable_column_value()->mutable_double_values()->mutable_values()->begin(),
        filter_eq->mutable_column_value()->mutable_double_values()->mutable_values()->end());
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    // contain
    filter_eq->set_column_name("floats");
    f1 = 0.985;
    filter_eq->mutable_column_value()->set_double_value(f1);
    filter_eq->set_action(hawking::util::ExpressionOp::CONTAIN);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    f1 = 0.987;
    filter_eq->mutable_column_value()->set_double_value(f1);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    // intersect
    f1 = 0.888;
    filter_eq->set_column_name("floats");
    filter_eq->mutable_column_value()->mutable_double_values()->add_values(f2);
    filter_eq->mutable_column_value()->mutable_double_values()->add_values(f1);
    filter_eq->set_action(hawking::util::ExpressionOp::INTERSECT);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    f1 = 0.987;
    filter_eq->mutable_column_value()->mutable_double_values()->add_values(f1);
    std::sort(
        filter_eq->mutable_column_value()->mutable_double_values()->mutable_values()->begin(),
        filter_eq->mutable_column_value()->mutable_double_values()->mutable_values()->end());
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

///////////////////////////////////////////////////////////////////////////////// string
    /// eq
    filter_eq->set_column_name("string");
    filter_eq->mutable_column_value()->set_bytes_value("hello world");
    filter_eq->set_action(hawking::util::ExpressionOp::EQ);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_bytes_value("helko world");
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);
    
    // ne
    filter_eq->mutable_column_value()->set_bytes_value("helko world");
    filter_eq->set_action(hawking::util::ExpressionOp::NE);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_bytes_value("hello world");
    filter_eq->set_action(hawking::util::ExpressionOp::NE);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    // >,<
    filter_eq->mutable_column_value()->set_bytes_value("hello world");
    filter_eq->set_action(hawking::util::ExpressionOp::LT);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->set_bytes_value("helmo world");
    filter_eq->set_action(hawking::util::ExpressionOp::LT);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_bytes_value("helko world");
    filter_eq->set_action(hawking::util::ExpressionOp::GT);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_bytes_value("helo world");
    filter_eq->set_action(hawking::util::ExpressionOp::GT);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    // >=,<=
    filter_eq->mutable_column_value()->set_bytes_value("hello world");
    filter_eq->set_action(hawking::util::ExpressionOp::LE);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_bytes_value("helmo world");
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_bytes_value("helko world");
    filter_eq->set_action(hawking::util::ExpressionOp::GE);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_bytes_value("hello world");
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    // in
    filter_eq->mutable_column_value()->mutable_bytes_values()->add_values("hella world");
    filter_eq->mutable_column_value()->mutable_bytes_values()->add_values("aaa");
    filter_eq->set_action(hawking::util::ExpressionOp::IN);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->mutable_bytes_values()->add_values("hello world");
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    // contain
    filter_eq->set_column_name("strings");
    filter_eq->mutable_column_value()->set_bytes_value("bbbbb");
    filter_eq->set_action(hawking::util::ExpressionOp::CONTAIN);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->set_bytes_value("aaaaa");
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    // intersect
    filter_eq->set_column_name("strings");
    filter_eq->mutable_column_value()->mutable_bytes_values()->add_values("bbbbb");
    filter_eq->mutable_column_value()->mutable_bytes_values()->add_values("ccccc");
    filter_eq->mutable_column_value()->mutable_bytes_values()->add_values("hella");
    filter_eq->set_action(hawking::util::ExpressionOp::INTERSECT);
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->mutable_bytes_values()->add_values("hello");
    ret = mem_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);
}

TEST(mem_segment_localdata_filter_test, add_and_select) {
    std::shared_ptr<IndexlibIndexConfig> index_config = MakeIndexConfig();
    MemSegment mem_segment;
    EXPECT_EQ(true, mem_segment.Init(index_config));
    
    std::shared_ptr<DocumentCreator> document_creator =
        std::make_shared<DocumentCreator>(index_config);

    // i8 filter test
    IndexlibMessage message;
    MakeAddMsg(&message);
    std::shared_ptr<NormalDocument> doc =
        document_creator->CreateNormalDocument(message, 0);
    if (!doc) {
        return;
    }
    doc->SetDocId(0);
    LOG(INFO) << mem_segment.AddDocument(doc);

    google::protobuf::RepeatedPtrField<hawking::util::SelectColumn> select_columns;
    hawking::util::SelectColumn* col_i8 = select_columns.Add();
    col_i8->set_column_name("i8");
    hawking::util::SelectColumn* col_i32 = select_columns.Add();
    col_i32->set_column_name("i32");
    hawking::util::SelectColumn* col_i64 = select_columns.Add();
    col_i64->set_column_name("i64");
    hawking::util::SelectColumn* col_float = select_columns.Add();
    col_float->set_column_name("float");
    hawking::util::SelectColumn* col_double = select_columns.Add();
    col_double->set_column_name("double");
    hawking::util::SelectColumn* col_string = select_columns.Add();
    col_string->set_column_name("string");
    hawking::util::SelectColumn* col_i8s = select_columns.Add();
    col_i8s->set_column_name("i8s");
    hawking::util::SelectColumn* col_i32s = select_columns.Add();
    col_i32s->set_column_name("i32s");
    hawking::util::SelectColumn* col_i64s = select_columns.Add();
    col_i64s->set_column_name("i64s");
    hawking::util::SelectColumn* col_floats = select_columns.Add();
    col_floats->set_column_name("floats");
    hawking::util::SelectColumn* col_doubles = select_columns.Add();
    col_doubles->set_column_name("doubles");
    hawking::util::SelectColumn* col_strings = select_columns.Add();
    col_strings->set_column_name("strings");

    mem_segment.FillLocaldata(10000000, &select_columns);
    for (const auto& col : select_columns) {
        LOG(INFO) << col.ShortDebugString();
    }
}

TEST(mem_segment_localdata_filter_test, dump_and_load) {
    std::shared_ptr<IndexlibIndexConfig> index_config = MakeIndexConfig();
    MemSegment mem_segment;
    EXPECT_EQ(true, mem_segment.Init(index_config));
    
    std::shared_ptr<DocumentCreator> document_creator =
        std::make_shared<DocumentCreator>(index_config);

    // i8 filter test
    IndexlibMessage message;
    MakeAddMsg(&message);
    std::shared_ptr<NormalDocument> doc =
        document_creator->CreateNormalDocument(message, 0);
    if (!doc) {
        return;
    }
    doc->SetDocId(0);
    LOG(INFO) << mem_segment.AddDocument(doc);

    mem_segment.Dump("/tmp/segment");

    DiskSegment disk_segment;
    disk_segment.Load(index_config, "/tmp/segment", FSFileType::FSFT_MMAP);

    google::protobuf::RepeatedPtrField<hawking::util::SelectColumn> select_columns;
    hawking::util::SelectColumn* col_i8 = select_columns.Add();
    col_i8->set_column_name("i8");
    hawking::util::SelectColumn* col_i32 = select_columns.Add();
    col_i32->set_column_name("i32");
    hawking::util::SelectColumn* col_i64 = select_columns.Add();
    col_i64->set_column_name("i64");
    hawking::util::SelectColumn* col_float = select_columns.Add();
    col_float->set_column_name("float");
    hawking::util::SelectColumn* col_double = select_columns.Add();
    col_double->set_column_name("double");
    hawking::util::SelectColumn* col_string = select_columns.Add();
    col_string->set_column_name("string");
    hawking::util::SelectColumn* col_i8s = select_columns.Add();
    col_i8s->set_column_name("i8s");
    hawking::util::SelectColumn* col_i32s = select_columns.Add();
    col_i32s->set_column_name("i32s");
    hawking::util::SelectColumn* col_i64s = select_columns.Add();
    col_i64s->set_column_name("i64s");
    hawking::util::SelectColumn* col_floats = select_columns.Add();
    col_floats->set_column_name("floats");
    hawking::util::SelectColumn* col_doubles = select_columns.Add();
    col_doubles->set_column_name("doubles");
    hawking::util::SelectColumn* col_strings = select_columns.Add();
    col_strings->set_column_name("strings");

    disk_segment.FillLocaldata(10000000, &select_columns);
    for (const auto& col : select_columns) {
        LOG(INFO) << col.ShortDebugString();
    }
}

TEST(disk_segment_localdata_filter_test, add_and_lookup_and_filter) {
    std::shared_ptr<IndexlibIndexConfig> index_config = MakeIndexConfig();
    MemSegment mem_segment;
    EXPECT_EQ(true, mem_segment.Init(index_config));
    
    std::shared_ptr<DocumentCreator> document_creator =
        std::make_shared<DocumentCreator>(index_config);

    // i8 filter test
    IndexlibMessage message;
    MakeAddMsg(&message);
    std::shared_ptr<NormalDocument> doc =
        document_creator->CreateNormalDocument(message, 0);
    if (!doc) {
        return;
    }
    doc->SetDocId(0);
    LOG(INFO) << mem_segment.AddDocument(doc);

    mem_segment.Dump("/tmp/segment");

    DiskSegment disk_segment;
    disk_segment.Load(index_config, "/tmp/segment", FSFileType::FSFT_MMAP);

    google::protobuf::RepeatedPtrField<hawking::util::FilterColumn> filter_columns;
    google::protobuf::RepeatedPtrField<hawking::util::FilterColumn> where_columns;

//////////////////////////////////////////////////////////////////////////////////// i8
    /// eq
    hawking::util::FilterColumn* filter_eq = filter_columns.Add();
    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(100);
    filter_eq->set_action(hawking::util::ExpressionOp::EQ);
    auto ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(101);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    /// ne
    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(100);
    filter_eq->set_action(hawking::util::ExpressionOp::NE);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(101);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    /// >,<
    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(99);
    filter_eq->set_action(hawking::util::ExpressionOp::LT);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(101);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(99);
    filter_eq->set_action(hawking::util::ExpressionOp::GT);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(101);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    /// >=,<=
    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(100);
    filter_eq->set_action(hawking::util::ExpressionOp::LE);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(99);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(100);
    filter_eq->set_action(hawking::util::ExpressionOp::GE);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(101);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    // in
    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->mutable_int_values()->add_values(99);
    filter_eq->mutable_column_value()->mutable_int_values()->add_values(101);
    filter_eq->set_action(hawking::util::ExpressionOp::IN);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->mutable_int_values()->add_values(100);
    std::sort(
        filter_eq->mutable_column_value()->mutable_int_values()->mutable_values()->begin(),
        filter_eq->mutable_column_value()->mutable_int_values()->mutable_values()->end());
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    // contain
    filter_eq->set_column_name("i8s");
    filter_eq->mutable_column_value()->set_int_value(5);
    filter_eq->set_action(hawking::util::ExpressionOp::CONTAIN);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->set_int_value(2);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    // intersect
    filter_eq->set_column_name("i8s");
    filter_eq->mutable_column_value()->mutable_int_values()->add_values(10);
    filter_eq->mutable_column_value()->mutable_int_values()->add_values(11);
    filter_eq->set_action(hawking::util::ExpressionOp::INTERSECT);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->mutable_int_values()->add_values(1);
    std::sort(
        filter_eq->mutable_column_value()->mutable_int_values()->mutable_values()->begin(),
        filter_eq->mutable_column_value()->mutable_int_values()->mutable_values()->end());
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

///////////////////////////////////////////////////////////////////////////////// float
    /// eq
    filter_eq->set_column_name("float");
    filter_eq->mutable_column_value()->set_double_value(0.123);
    filter_eq->set_action(hawking::util::ExpressionOp::EQ);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_double_value(0.124);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    // ne
    filter_eq->mutable_column_value()->set_double_value(0.123);
    filter_eq->set_action(hawking::util::ExpressionOp::NE);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->set_double_value(0.124);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    // >,<
    filter_eq->mutable_column_value()->set_double_value(0.123);
    filter_eq->set_action(hawking::util::ExpressionOp::LT);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->set_double_value(0.124);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_double_value(0.123);
    filter_eq->set_action(hawking::util::ExpressionOp::GT);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->set_double_value(0.122);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    // >=,<=
    filter_eq->mutable_column_value()->set_double_value(0.123);
    filter_eq->set_action(hawking::util::ExpressionOp::LE);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_double_value(0.122);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->set_double_value(0.123);
    filter_eq->set_action(hawking::util::ExpressionOp::GE);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_double_value(0.124);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    // in
    float f1 = 0.122, f2 = 0.124, f3 = 0.123;
    filter_eq->mutable_column_value()->mutable_double_values()->add_values(f1);
    filter_eq->mutable_column_value()->mutable_double_values()->add_values(f2);
    filter_eq->set_action(hawking::util::ExpressionOp::IN);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->mutable_double_values()->add_values(f3);
    std::sort(
        filter_eq->mutable_column_value()->mutable_double_values()->mutable_values()->begin(),
        filter_eq->mutable_column_value()->mutable_double_values()->mutable_values()->end());
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    // contain
    filter_eq->set_column_name("floats");
    f1 = 0.985;
    filter_eq->mutable_column_value()->set_double_value(f1);
    filter_eq->set_action(hawking::util::ExpressionOp::CONTAIN);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    f1 = 0.987;
    filter_eq->mutable_column_value()->set_double_value(f1);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    // intersect
    f1 = 0.888;
    filter_eq->set_column_name("floats");
    filter_eq->mutable_column_value()->mutable_double_values()->add_values(f2);
    filter_eq->mutable_column_value()->mutable_double_values()->add_values(f1);
    filter_eq->set_action(hawking::util::ExpressionOp::INTERSECT);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    f1 = 0.987;
    filter_eq->mutable_column_value()->mutable_double_values()->add_values(f1);
    std::sort(
        filter_eq->mutable_column_value()->mutable_double_values()->mutable_values()->begin(),
        filter_eq->mutable_column_value()->mutable_double_values()->mutable_values()->end());
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

///////////////////////////////////////////////////////////////////////////////// string
    /// eq
    filter_eq->set_column_name("string");
    filter_eq->mutable_column_value()->set_bytes_value("hello world");
    filter_eq->set_action(hawking::util::ExpressionOp::EQ);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_bytes_value("helko world");
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);
    
    // ne
    filter_eq->mutable_column_value()->set_bytes_value("helko world");
    filter_eq->set_action(hawking::util::ExpressionOp::NE);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_bytes_value("hello world");
    filter_eq->set_action(hawking::util::ExpressionOp::NE);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    // >,<
    filter_eq->mutable_column_value()->set_bytes_value("hello world");
    filter_eq->set_action(hawking::util::ExpressionOp::LT);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->set_bytes_value("helmo world");
    filter_eq->set_action(hawking::util::ExpressionOp::LT);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_bytes_value("helko world");
    filter_eq->set_action(hawking::util::ExpressionOp::GT);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_bytes_value("helo world");
    filter_eq->set_action(hawking::util::ExpressionOp::GT);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    // >=,<=
    filter_eq->mutable_column_value()->set_bytes_value("hello world");
    filter_eq->set_action(hawking::util::ExpressionOp::LE);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_bytes_value("helmo world");
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_bytes_value("helko world");
    filter_eq->set_action(hawking::util::ExpressionOp::GE);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    filter_eq->mutable_column_value()->set_bytes_value("hello world");
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    // in
    filter_eq->mutable_column_value()->mutable_bytes_values()->add_values("hella world");
    filter_eq->mutable_column_value()->mutable_bytes_values()->add_values("aaa");
    filter_eq->set_action(hawking::util::ExpressionOp::IN);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->mutable_bytes_values()->add_values("hello world");
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    // contain
    filter_eq->set_column_name("strings");
    filter_eq->mutable_column_value()->set_bytes_value("bbbbb");
    filter_eq->set_action(hawking::util::ExpressionOp::CONTAIN);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->set_bytes_value("aaaaa");
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);

    // intersect
    filter_eq->set_column_name("strings");
    filter_eq->mutable_column_value()->mutable_bytes_values()->add_values("bbbbb");
    filter_eq->mutable_column_value()->mutable_bytes_values()->add_values("ccccc");
    filter_eq->mutable_column_value()->mutable_bytes_values()->add_values("hella");
    filter_eq->set_action(hawking::util::ExpressionOp::INTERSECT);
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), false);

    filter_eq->mutable_column_value()->mutable_bytes_values()->add_values("hello");
    ret = disk_segment.FilterCheck(10000000, filter_columns, where_columns);
    EXPECT_EQ(ret.Code(), Status::OK);
    EXPECT_EQ(ret.Value(), true);
}

int main(int argc, char* argv[]) {
    ::testing::InitGoogleTest(&argc, argv);
    ::google::ParseCommandLineFlags(&argc, &argv, true);
    srand(static_cast<unsigned>(time(0)));
    return RUN_ALL_TESTS();
}